package com.lagou.registry;

import com.lagou.consumer.ConsumerNettyConnect;
import com.lagou.registry.listener.ConsumerListener;
import com.lagou.consumer.ConsumerBeanPostProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

@Component
@DependsOn("consumerBeanPostProcessor")
public class ZkConsumerRegistry extends ZkRegistry {

    @Autowired
    private ZkProviderRegistry providerRegistry;

    @Autowired
    private ConsumerNettyConnect nettyConnect;

    @Override
    public String getPath() {
        return "/consumer";
    }

    //启动时，
    @Override
    @PostConstruct
    public void init() {
        //遍历当前项目，所有的消费者信息
        Set<Class> consumerClassSet = ConsumerBeanPostProcessor.classSet;
        if(consumerClassSet != null && consumerClassSet.size() > 0){
            List<RpcProviderNode> listNode = providerRegistry.getAllRpcProviderNode();
            List<RpcProviderNode> needList = new ArrayList<>();
            //查找当前项目中，需要用到的RPC服务，并将他们存储到list中
            for (Class clazz : consumerClassSet) {
                for (RpcProviderNode rpcProviderNode : listNode) {
                    Set<String> appClassSet = rpcProviderNode.getClassName();
                    if (appClassSet.contains(clazz.getName())) {
                        needList.add(rpcProviderNode);
                    }
                }
            }
            //遍历需要用到的providerNode服务器信息，连接服务器
            for (RpcProviderNode rpcProviderNode : needList) {
                String ip = rpcProviderNode.getIp();
                int port = rpcProviderNode.getPort();
                nettyConnect.conn(ip,port);
            }
            //启动一个线程去监听生产者，当服务器节点信息发生变化时，自动路由处理
            ConsumerListener listener = new ConsumerListener(providerRegistry.getCurrentNamespace(),nettyConnect,providerRegistry);
            Thread thread = new Thread(listener);
            thread.setDaemon(true);
            thread.start();
        }

    }
    @Override
    public Class<RpcConsumerNode> getServerNodeClass() {
        return RpcConsumerNode.class;
    }
}
